package sc.analysis.vortex;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;

import sc.writable.IntermediateData;

/**
 * MapReduce job to get a score for Vortex technical analysis.
 * 
 * This indicator shows if a new trend (bull or bear) is forming.
 * 
 * @author Quentin Rossé
 */
public class Vortex {

	/**
	 * @param args inputDirectory outputDirectory
	 * 
	 * @throws ClassNotFoundException 
	 * @throws InterruptedException 
	 * @throws IOException 
	 */
	public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
		
		Job job = new Job();
		
		Configuration conf = job.getConfiguration();
		
		job.setJobName("Vortex Indicator");
		job.setJarByClass(Vortex.class);
		
		FileSystem fileSystem = FileSystem.get(conf);
		
		//Overwrites the output directory if it exists
	    Path path = new Path(args[1]);
	    if (fileSystem.exists(path)) {
	    	fileSystem.delete(path, true);
	    }
	    
	    conf.set("inputDirectory", args[0]);
	    conf.set("outputDirectory", args[1]);
	    
		job.setInputFormatClass(SequenceFileInputFormat.class);
		
		job.setMapperClass(VortexMapper.class);
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(VortexData.class);
		
		job.setNumReduceTasks(1);
		job.setReducerClass(VortexReducer.class);
		
		job.setOutputFormatClass(SequenceFileOutputFormat.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntermediateData.class);
		
		
		FileInputFormat.setInputPaths(job, args[0]);
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        
		job.submit();
		job.waitForCompletion(true);
	}

}
